/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package service

import (
	"context"
	"sync"

	"hundsun.com/hsl/hschain/common/config"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/types"
	"hundsun.com/hsl/hschain/p2p/common"
	"hundsun.com/hsl/hschain/p2p/common/p2perror"
	"hundsun.com/hsl/hschain/p2p/impl/dht/node"
	"hundsun.com/hsl/hschain/p2p/impl/dht/protocol"
	"hundsun.com/hsl/hschain/p2p/impl/dht/txpool"
	"hundsun.com/hsl/hschain/store"
	dbm "hundsun.com/hsl/hschain/store/db"
	"hundsun.com/hsl/hschain/store/mq"
)

//P2p P2p
type P2p struct {
	logging.Log
	Mconfig      *config.ModuleConfig
	Cfg          *common.P2pConfig
	Ctx          context.Context
	CancelFun    context.CancelFunc
	TopicClient  mq.Client  // topic类型client
	WorkerClient mq.Client  // worker类型client
	P2pNode      *node.Node // p2p节点
	TaskWg       *sync.WaitGroup
	P2pDB        dbm.Database         // 这里用于存储p2p节点信息的数据库
	TxPool       txpool.PendingTxPool //用于对接txpool
}

//NewP2p 新建p2p
func NewP2p(ctx context.Context, cfg *config.ModuleConfig, cnlfn context.CancelFunc, logger logging.Log) *P2p {
	tClient, err := store.NewMqClient(mq.ModeTopic.String())
	if err != nil {
		panic(err)
	}
	tClient.Sub(types.P2pModule, mq.ModeTopic.String())

	wClient, err := store.NewMqClient(mq.ModeWorker.String())
	if err != nil {
		panic(err)
	}
	wClient.Sub(types.P2pModule, mq.ModeWorker.String())

	p := &P2p{
		Log:          logger,
		Mconfig:      cfg,
		Cfg:          common.InitP2pConfig(cfg),
		Ctx:          ctx,
		CancelFun:    cnlfn,
		TopicClient:  tClient,
		WorkerClient: wClient,
		TaskWg:       &sync.WaitGroup{},
		TxPool:       txpool.NewTxPool(ctx, logger),
		P2pDB:        store.GetP2pDB(),
		// Database赋值
	}
	return p
}

//Init 初始化p2p host相关资源
func (p *P2p) Init() error {
	if p.Cfg == nil {
		p.Cfg = common.InitP2pConfig(p.Mconfig)
	}

	n, err := node.Init(p.Ctx, p.Cfg, p.Log, p.P2pDB)
	if err != nil {
		panic(err)
	}
	p.P2pNode = n
	return nil
}

//StartP2p 启动p2p
func (p *P2p) StartP2p() {
	// 获取ledger句柄
	p2perror.Init(p.Log)

	env := &protocol.P2PEnv{
		Log:         p.Log,
		Cfg:         p.Cfg,
		Ctx:         p.Ctx,
		TopicClient: p.TopicClient,
		WorkClient:  p.WorkerClient,
		P2pNode:     p.P2pNode,
		TxPool:      p.TxPool,
	}
	protocol.InitProtocols(env)

	// node start
	err := p.P2pNode.Start()
	if err != nil {
		panic(err)
	}

	// 处理消息事件
	go p.handleTopicEvent()
	go p.handleWorkerEvent()
}

// CloseP2p 关闭p2p释放资源
func (p *P2p) CloseP2p() {
	p.Log.Info("p2p closing")
	if p.Ctx != nil {
		p.Ctx.Done()
	}
	if p.TopicClient != nil {
		p.TopicClient.Close()
	}
	if p.WorkerClient != nil {
		p.WorkerClient.Close()
	}
	if p.P2pNode != nil {
		p.P2pNode.Close()
	}
	p.P2pDB.Close()
	if p.TaskWg != nil {
		p.TaskWg.Wait()
	}
	p.Log.Info("p2p closed")
}

func (p *P2p) handleTopicEvent() {
	// 开始处理Event
	for msg := range p.TopicClient.Recv() {
		if handler := protocol.GetEventHandler(msg.Ty); handler != nil {
			p.TaskWg.Add(1)
			go func(m *mq.Message) {
				defer p.TaskWg.Done()
				handler(m)
			}(msg)
		} else {
			p.Log.Error("handleTopicEvent", "unknown message type", msg.Ty)
		}
	}
}

func (p *P2p) handleWorkerEvent() {
	// 开始处理Event
	for msg := range p.WorkerClient.Recv() {
		if handler := protocol.GetEventHandler(msg.Ty); handler != nil {
			p.TaskWg.Add(1)
			go func(m *mq.Message) {
				defer p.TaskWg.Done()
				handler(m)
			}(msg)
		} else {
			p.Log.Error("handleWorkerEvent", "unknown message type", msg.Ty)
		}
	}
}
